package com.xian.java.sink;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class MyFileSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> ds = env.readTextFile("J:\\test\\wc.txt");

      /*  DataStream<Tuple2<String, Integer>> ds2 = ds.flatMap(new RichFlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] arr = line.split(" ");
                for (String word : arr) {
                    Tuple2<String, Integer> tuple2 = new Tuple2<>(word, 1);
                    collector.collect(tuple2);
                }
            }
        });*/

        DataStream<String> ds2 = ds.flatMap(new RichFlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
            }
        });




        DataStream<Tuple2<String, Integer>> ds3 = ds2.map(new RichMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = ds3.keyBy(0);
        DataStream<Tuple2<String, Integer>> ds4 = keyedStream.sum(1);

        ds4.print();

        ds4.writeAsText("J:\\test\\out1");


        env.execute();
    }
}
